package com.raven.play.stream.test.javadsl.tutorial;

import akka.Done;
import akka.NotUsed;
import akka.stream.FlowShape;
import akka.stream.SinkShape;
import akka.stream.SourceShape;
import akka.stream.UniformFanOutShape;
import akka.stream.javadsl.*;
import com.raven.play.stream.test.BaseTest;
import org.junit.jupiter.api.Test;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletionStage;

public class $12GraphMaterializedValues extends BaseTest {

    @Test
    public void test1(){
        Source<String, NotUsed> source = Source.from(List.of("Akka", "is", "awesome", "rock", "the", "jvm"));

        Sink<String, CompletionStage<Done>> printer = Sink.foreach(it -> System.out.println("it = " + it));
        Sink<String, CompletionStage<Integer>> counter = Sink.fold(0, (count, s) -> count + 1);
        Flow<String, String, NotUsed> lowercaseFilter = Flow.of(String.class).filter(word -> Objects.equals(word.toLowerCase(), word));
        Flow<String, String, NotUsed> shortStringFilter = Flow.of(String.class).filter(word -> word.length() < 5);

        @SuppressWarnings("unchecked")
        Sink<String, NotUsed> complexWordSink = Sink.fromGraph(GraphDSL.create(builder -> {
            SinkShape<String> printerShape = builder.add(printer);
            SinkShape<String> counterShape = builder.add(counter);
            FlowShape<String, String> lowercaseFilterShape = builder.add(lowercaseFilter);
            FlowShape<String, String> shortStringFilterShape = builder.add(shortStringFilter);

            UniformFanOutShape<String, String> broadcast = builder.add(Broadcast.create(2));

            builder.from(broadcast.out(0)).via(lowercaseFilterShape).to(printerShape);
            builder.from(broadcast.out(1)).via(shortStringFilterShape).to(counterShape);

            return SinkShape.of(broadcast.in());
        }));

        // 这种方式下, 无法控制 materialized 值
        NotUsed run = source.toMat(complexWordSink, Keep.right()).run(system);

    }

    @Test
    public void test2(){
        Source<String, NotUsed> source = Source.from(List.of("Akka", "is", "awesome", "rock", "the", "jvm"));

        Sink<String, CompletionStage<Done>> printer = Sink.foreach(it -> System.out.println("it = " + it));
        Sink<String, CompletionStage<Integer>> counter = Sink.fold(0, (count, s) -> count + 1);
        Flow<String, String, NotUsed> lowercaseFilter = Flow.of(String.class).filter(word -> Objects.equals(word.toLowerCase(), word));
        Flow<String, String, NotUsed> shortStringFilter = Flow.of(String.class).filter(word -> word.length() < 5);

        @SuppressWarnings("unchecked")
        Sink<String, CompletionStage<Integer>> complexWordSink = Sink.fromGraph(GraphDSL.create(counter, (builder, counterShape) -> {
            // SourceShape<String> sourceShape = builder.add(source);
            SinkShape<String> printerShape = builder.add(printer);
            // SinkShape<Integer> counterShape = builder.add(counter);
            FlowShape<String, String> lowercaseFilterShape = builder.add(lowercaseFilter);
            FlowShape<String, String> shortStringFilterShape = builder.add(shortStringFilter);

            UniformFanOutShape<String, String> broadcast = builder.add(Broadcast.create(2));

            builder.from(broadcast.out(0)).via(lowercaseFilterShape).to(printerShape);
            builder.from(broadcast.out(1)).via(shortStringFilterShape).to(counterShape);

            return SinkShape.of(broadcast.in());
        }));

        CompletionStage<Integer> future = source.toMat(complexWordSink, Keep.right()).run(system);
        future.whenComplete((count, e) ->{
            if (e != null) {
                log.error(e.getMessage(), e);
            }else{
                log.info("Total number is: {}", count);
            }
        });
    }

    @Test
    public void test3(){
        Source<String, NotUsed> source = Source.from(List.of("Akka", "is", "awesome", "rock", "the", "jvm"));

        Sink<String, CompletionStage<Done>> printer = Sink.foreach(it -> System.out.println("it = " + it));
        Sink<String, CompletionStage<Integer>> counter = Sink.fold(0, (count, s) -> count + 1);
        Flow<String, String, NotUsed> lowercaseFilter = Flow.of(String.class).filter(word -> Objects.equals(word.toLowerCase(), word));
        Flow<String, String, NotUsed> shortStringFilter = Flow.of(String.class).filter(word -> word.length() < 5);

        @SuppressWarnings("unchecked")
        Sink<String, CompletionStage<Integer>> complexWordSink = Sink.fromGraph(GraphDSL.create(printer, counter, Keep.right(), (builder, printerShape, counterShape) -> {
            FlowShape<String, String> lowercaseFilterShape = builder.add(lowercaseFilter);
            FlowShape<String, String> shortStringFilterShape = builder.add(shortStringFilter);

            UniformFanOutShape<String, String> broadcast = builder.add(Broadcast.create(2));

            builder.from(broadcast.out(0)).via(lowercaseFilterShape).to(printerShape);
            builder.from(broadcast.out(1)).via(shortStringFilterShape).to(counterShape);


            return SinkShape.of(broadcast.in());
        }));

        CompletionStage<Integer> future = source.toMat(complexWordSink, Keep.right()).run(system);
        future.whenComplete((count, e) ->{
            if (e != null) {
                log.error(e.getMessage(), e);
            }else{
                log.info("Total number is: {}", count);
            }
        });
    }
}
